package com.tbit.uqbike.service.base.impl;

import com.tbit.uqbike.service.base.RedisService;
import com.tbit.uqbike.webmanager.util.redis.RedisUtil;
import com.tbit.utils.DateTimeUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Service("redisService")
@Slf4j
public class RedisServiceImpl implements RedisService {
    @Resource
    @Qualifier("redisTemplate")
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private RedisUtil redisUtil;

    private static  int BATCH_SIZE = 10000;

    @Override
    public void add(String key, String value) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        valueOperation.set(key, value);
    }

    @Override
    public String get(String key) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();

        return valueOperation.get(key);
    }

    @Override
    public void del(String key) {
        redisTemplate.delete(key);
    }

    @Override
    public String get(String hashId, String key) {
        HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash();
        return hashOperations.get(hashId, key);
    }

    @Override
    public Map<String, String> getAll(String hashId) {
        HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash();
        return hashOperations.entries(hashId);
    }

    @Override
    public void incr(String hashId) {
        if (!hasKey(DateTimeUtil.getNowTimeYMD() + "." + hashId)) {
            redisUtil.set(DateTimeUtil.getNowTimeYMD() + "." + hashId, "1");
            return;
        } else {
            int count = Integer.parseInt(redisUtil.get(DateTimeUtil.getNowTimeYMD() + "." + hashId)) + 1;
            redisUtil.set(DateTimeUtil.getNowTimeYMD() + "." + hashId, "" + count);
        }

    }

    @Override
    public void incrByRegionId(String hashId) {
        if (!hasKey(DateTimeUtil.getNowTimeYMD() + "_" + hashId)) {
            redisUtil.set(DateTimeUtil.getNowTimeYMD() + "_" + hashId, "1");
            return;
        } else {
            int count = Integer.parseInt(redisUtil.get(DateTimeUtil.getNowTimeYMD() + "_" + hashId)) + 1;
            redisUtil.set(DateTimeUtil.getNowTimeYMD() + "_" + hashId, "" + count);
        }
    }

    @Override
    public Boolean hasKey(String hashId) {
        return redisUtil.hasKey(hashId);
    }

    @Override
    public void add(String key, String value, Integer seconds) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        if (seconds == -1) {
            valueOperation.set(key, value);
        } else {
            valueOperation.set(key, value, seconds, TimeUnit.SECONDS);
        }
    }

    @Override
    public void add(String key, String value, long timeout, TimeUnit unit) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        valueOperation.set(key, value, timeout, unit);
    }

    @Override
    public void delByPrex(String prex) {
        Set<String> keys = redisTemplate.keys(prex + "*");
        if (!CollectionUtils.isEmpty(keys)) {
            redisTemplate.delete(keys);
        }
    }

    /**
     * 通过前缀批量删除key
     *
     * @param prefix key prefix
     */
    @Override
    public void delByPrefixPipeline(String prefix) {
        log.info("Batch deleting keys with prefix '{}'", prefix);
        ScanOptions scanOptions = ScanOptions.scanOptions()
                .match(prefix + "*")
                .count(BATCH_SIZE)
                .build();

        long startTime = System.currentTimeMillis();
        AtomicInteger total = new AtomicInteger();

        Cursor<byte[]> cursor = redisTemplate.executeWithStickyConnection(redisConnection -> redisConnection.scan(scanOptions));
        List<byte[]> keysToDelete = new ArrayList<>();
        while (cursor.hasNext()) {
            keysToDelete.add(cursor.next());
            if (keysToDelete.size() >= BATCH_SIZE) {
                deleteKeys(keysToDelete, total);
                keysToDelete.clear();
            }
        }
        deleteKeys(keysToDelete, total);

        try {
            cursor.close();
        } catch (Exception e) {
            log.error("Error occurred while closing cursor", e);
        }

        long endTime = System.currentTimeMillis();
        log.info("Deleted {} keys with prefix '{}' in {} milliseconds", total.get(), prefix, endTime - startTime);
    }

    /**
     * 批量删除key
     *
     * @param keysToDelete keys to delete
     * @param total        总数，用于统计
     */

    private void deleteKeys(List<byte[]> keysToDelete, AtomicInteger total) {
        if (!keysToDelete.isEmpty()) {
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                connection.del(keysToDelete.toArray(new byte[keysToDelete.size()][]));
                total.addAndGet(keysToDelete.size());
                return null;
            });
        }
    }



    @Override
    public boolean locked(String key, Integer second) {
        if (hasKey(key)) {
            return true;
        }
        add(key, "1", second, TimeUnit.SECONDS);
        return false;
    }

    @Override
    public List<String> getList(String key) {
        List<String> values = null;
        try {
            ListOperations<String, String> listOpreations = redisTemplate.opsForList();
            values = listOpreations.range(key, 0, -1);
        } catch (SerializationException e) {
            del(key);
        }
        return values;
    }

    @Override
    public void delToList(String key, String value) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        listOperations.remove(key, -1, value);
    }

    @Override
    public boolean setNx(String key, String value, long liveTime) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        boolean result = valueOperation.setIfAbsent(key, value, liveTime, TimeUnit.SECONDS);
        return result;
    }

    @Override
    public void addSet(String key, Date expireDate, String... values) {
        redisTemplate.opsForSet().add(key,values);
        redisTemplate.expireAt(key,expireDate);
    }

    @Override
    public boolean hasElementSet(String key, String value) {
        return Boolean.TRUE.equals(redisTemplate.opsForSet().isMember(key, value));
    }

    /**
     * 删除hash表中的值
     * @param key  键 不能为null
     * @param item 项 可以使多个 不能为null
     */
    public void hdel(String key, Object... item) {
        redisTemplate.opsForHash().delete(key, item);
    }

    @Override
    public Long incr(String key, long liveTime) {
        RedisAtomicLong entityIdCounter = new RedisAtomicLong(key, Objects.requireNonNull(redisTemplate.getConnectionFactory()));
        Long increment = entityIdCounter.getAndIncrement();
        if (Objects.equals(increment, 0L) && liveTime > 0) {
            entityIdCounter.expire(liveTime, TimeUnit.SECONDS);
        }
        return increment;
    }
}
